package Client.serviceCenter;

import Client.cache.serviceCache;
import Client.serviceCenter.ZkWatcher.watchZk;
import Client.serviceCenter.balance.impl.ConsistencyHashBalance;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.net.InetSocketAddress;
import java.util.List;

public class ZKServiceCenter implements ServiceCenter {

    // curator提供的zookeeper客户端
    private CuratorFramework zkClient;
    // zookeeper根路径节点：命令空间
    private static final String ZOOKEEPER_ROOT_PATH = "MyRPC";
    private static final String ZOOKEEPER_RETRY_PATH = "CanRetry";
    // 服务缓存字段
    private serviceCache serviceCache;

    // zkClient初始化,并于zookeeper服务端连接
    public ZKServiceCenter() throws InterruptedException {
        // 重试策略：指数时间重试,3次重试
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        /**
         * zookeeper地址固定,seesionTimeoutMs 与zoo.cfg 中的tickTime保持一致
         * zk还会根据minSessionTimeoutMs与maxSessionTimeoutMs来判断session是否超时
         * 使用心跳监听状态
         */
        this.zkClient = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(40000)
                .retryPolicy(retryPolicy)
                .namespace(ZOOKEEPER_ROOT_PATH)
                .build();
        this.zkClient.start();
        System.out.println("zookeeper连接成功");
        // 初始化缓存
        this.serviceCache = new serviceCache();
        // 加入zookeeper事件监听器
        watchZk watchZk = new watchZk(zkClient, serviceCache);
        // 监听根目录节点变化
        watchZk.watchToUpdate(ZOOKEEPER_ROOT_PATH);
    }

    // 根据服务名获取服务端地址
    @Override
    public InetSocketAddress serviceDiscover(String serviceName) {
        try {
            // 先从本地缓存中查找，如果没有则获取
            List<String> serviceList = serviceCache.getServiceFromCache(serviceName);
            // 获取服务名对应的路径下的所有子节点，格式：/MyPRC/com.example.rpc.UserService/127.0.0.1:8080
            // List<String> childNodes = zkClient.getChildren().forPath("/" + serviceName);
            if (serviceList == null) {
                serviceList = zkClient.getChildren().forPath("/" + serviceName);
            }
            // 打印服务端地址
            System.out.println("服务名:" + serviceName + " 服务端地址列表为：" + serviceList);
            // 负载均衡得到地址
            String address = new ConsistencyHashBalance().balance(serviceList);
            return parseAddress(address);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean checkRetry(String serviceName) {
        boolean canRetry = false;
        try {
            List<String> serviceList = zkClient.getChildren().forPath("/" + ZOOKEEPER_RETRY_PATH);
            for (String service : serviceList) {
                if (service.equals(serviceName)) {
                    System.out.println("服务名:" + serviceName + " 在白名单上，可重试");
                    canRetry = true;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return canRetry;
    }

    // 地址-> XXX.XXX.XXX.XXX:port 字符串
    private String getServiceAddress(InetSocketAddress address) {
        return address.getHostName() + ":" + address.getPort();
    }

    // 字符串解析为地址
    private InetSocketAddress parseAddress(String address) {
        String[] addressArray = address.split(":");
        return new InetSocketAddress(addressArray[0], Integer.parseInt(addressArray[1]));
    }
}
